package com.avic.szzx.consumer;

import com.alibaba.fastjson.JSONObject;
import com.avic.szzx.service.NebulaService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.LinkedHashMap;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
@RocketMQMessageListener(
//        topic = "ORDER_TOPIC",
        topic = "test_edge",
        consumerGroup = "order-consumer-group",
        consumeMode = ConsumeMode.ORDERLY,  // 必须设置为顺序模式[4,5](@ref)
        messageModel = MessageModel.CLUSTERING,
        selectorExpression = "*",
        consumeThreadNumber = 4  // 线程数必须 ≤ 队列数[4,5](@ref)
)
public class RocketConsumerEdge implements RocketMQListener<MessageExt> {
    @Autowired
    private NebulaService nebulaService;

    @Override
    public void onMessage(MessageExt message) {
        log.info("#########################RocketMQ消费者接收到消息：" + message);
        boolean result = false;

        try {
//            LinkedHashMap<String, Object> mapObj = JSONObject.parseObject(message, LinkedHashMap.class);
//            String tag = mapObj.get("tag").toString();
//            String option = mapObj.get("option").toString();
//            mapObj.remove("tag");
//            mapObj.remove("option");
            String body = new String(message.getBody(), "UTF-8");
            log.info("#########################RocketMQ消费者接收到顶点消息：" + body);
            System.out.println("#########################RocketMQ消费者接收到顶点消息：" + body);
            LinkedHashMap<String, Object> mapObj = JSONObject.parseObject(body, LinkedHashMap.class);
            String tag = mapObj.get("tag").toString();
            String option = mapObj.get("option").toString();
            mapObj.remove("tag");
            mapObj.remove("option");

            if ("ADD".equals(option)) {
                result = nebulaService.addEdgeData(mapObj, tag);
                if (!result) {
                    log.error("#############################图数据库：" + tag + "执行失败，数据回滚！,id为：[{}]", mapObj.get("id"));
                }
            } else if ("DELETE".equals(option)) {
                result = nebulaService.deleteEdgeData(mapObj, tag);
                if (!result) {
                    log.error("#############################图数据库：" + tag + "执行失败,id为：[{}]", mapObj.get("id"));
                }
            }

            if (result) {
                log.info("#############################图数据库：" + tag + "执行成功,id为：[{}]", mapObj.get("id"));
            }
        } catch (Exception e) {
            log.error("#############################消息处理异常：", e);
        }
    }

    private void processOrder(String orderId, String operation) {
        // 模拟业务处理
        try {
            Thread.sleep(100); // 模拟处理耗时
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.printf("订单 %s 处理完成: %s%n", orderId, operation);
    }
}